package com.wsx.gradeanaly;

import java.io.IOException;

import org.apache.hadoop.io.FloatWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class TestReducer extends Reducer<Text, Student, Text, Student> {

	private float max = Float.MIN_VALUE;
	private Student maxstu;
	private float min = Float.MAX_VALUE;
	private Student minstu;
	
	private double sum = 0;
	private long num = 0;

	@Override
	protected void reduce(Text key, Iterable<Student> values, Reducer<Text, Student, Text, Student>.Context context)
			throws IOException, InterruptedException {
		if ("max".equals(key.toString())) {
			for (Student stu : values) {
				if (stu.getScore() > max) {
					max = stu.getScore();
					maxstu = (Student)stu.clone();
				}
			}
		}else if ("min".equals(key.toString())) {
			for (Student stu : values) {
				if (stu.getScore() < min) {
					min = stu.getScore();
					minstu = (Student)stu.clone();
				}
			}
		}else if ("common".equals(key.toString())) {
			for (Student stu : values) {
				context.write(new Text("common"), stu);
			}
		}else if ("aver".equals(key.toString())) {
			for (Student stu : values) {
				String[] strs = stu.getName().split(";");
				sum += Double.parseDouble(strs[0]);
				num += Integer.parseInt(strs[1]);
			}
		}
	}

	@Override
	protected void cleanup(Reducer<Text, Student, Text, Student>.Context context)
			throws IOException, InterruptedException {
		context.write(new Text("max"), maxstu);
		context.write(new Text("min"), minstu);
		context.write(new Text("aver"), new Student("", (float)(sum/num), 'F'));
	}

}
